/*
 * MIT License
 *
 * Copyright (c) 2020 wen.gu <454727014@qq.com>
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 */

 /***************************************************************************
 * Name: ipc_service_provider.cpp
 *
 * Purpose: ipc service side implementation for 'COM' Functional.
 *
 * Developer:
 *   wen.gu , 2020-07-29
 *
 * TODO:
 *
 ***************************************************************************/

 /******************************************************************************
 **    INCLUDES
 ******************************************************************************/
#define LOG_TAG "ISPR"
#include "panda/sys/ipc_service_provider.h"

#include <set>
#include <map>

#include "panda/core/function_executor.h"
#include "panda/sys/ipc_packet.h"
#include "panda/sys/ipc_service.h"

#include "panda/core/log.h"

namespace panda
{
namespace sys
{
/******************************************************************************
 **    MACROS
 ******************************************************************************/

 struct IpcServiceClientInfo
{
    int32_t clientId = IPC_INVALID_ID;
    IpcPacketInfo pktInfo;
};


 struct IpcServiceRequestHandle
 {
     int32_t clientId = IPC_INVALID_ID;
     uint32_t sessionId = 0;
     std::string name = "";
 };

 //using IpcServiceClientInfoPtr = std::shared_ptr<IpcServiceClientInfo>;
/******************************************************************************
 **    VARIABLE DEFINITIONS
 ******************************************************************************/

class IpcServiceProvider::impl
{
public:
    using ServiceMethodMap = std::map<std::string, IpcServiceMethodHandler>;

    //using ServiceMsgQueue = std::queue< MessagePtr >;
    using ServiceAutoLock = std::unique_lock<std::mutex>;
    using GuardAutoLock = std::lock_guard<std::mutex>;

    using EventSubscribers = std::set<int32_t>;
    using FunctionExecutor = panda::core::FunctionExecutor;

    /** store event as 1 backup,
     * std::string: event name,
     * MessagePtr: the message of current event.
     */

    using EventBackup = std::map<std::string, MessagePtr>;


    /**std::string: event name,
     * EventSubscribers: client set, which subscribe current event.
     */
    using ServiceEventMap = std::map<std::string, EventSubscribers>;


   // using ComServiceReplyHandler = std::function<void(MessagePtr)>;
    /** map: packet id --> reply handler */
   // using ServiceReplyMap = std::map<uint32_t, ComServiceReplyHandler>;

    using IpcService = panda::sys::IpcService;

    /**map: clientId --> client info */
    using ServiceClientMap = std::map<int32_t, IpcServiceClientInfo>;
public:
    bool mIsStarted = false;
    bool mIsTerminate = false;
    uint32_t mNextPacketId = 0;
   
    IpcService* mService;
    IpcDataParser* mParser;


    std::string mSrvName = "";


    FunctionExecutor mExecutor;

    ServiceClientMap mClients;
    ServiceEventMap mEvents;
    EventBackup mEventBackup;

    std::mutex mReplyLock; /** mutex lock for the reply of callback */
    std::mutex mClientLock; /** mutex lock for process client info */
    std::mutex mApiLock; /** mutex lock for API */
    std::mutex mEventLock; /** mutex lock for process event */
    std::mutex mEvtRegLock; /** mutex lock for event register */
    std::mutex mGeneratePacketIdLock;
    ServiceMethodMap mMethods;
    MessageQueue mMsgQueue;

public:
    impl(const std::string& srvName, uint32_t multiThreadCnt);
    PandaErrc start();
    PandaErrc stop();
    PandaErrc registerMethod(const std::string& name, IpcServiceMethodHandler handler);

    PandaErrc sendResponse(IpcServiceRequestHandlePtr reqHdl, const uint8_t* data, uint32_t size);

    void emitSignal(const std::string& sigName, const uint8_t* data, uint32_t size);
    //void emitSignal(const std::string& sigName, MessagePtr msg);

protected:

    uint32_t generatePacketId();
    PandaErrc sendPacket(int32_t clientId, const std::string& name, uint32_t sessionId, MessageType type, const uint8_t* param, uint32_t size);
    PandaErrc sendMessage(int32_t clientId, MessagePtr msg);
protected:
    void onClientStateChange(int32_t clientId, bool connected);
    bool onProcessMessage();
    void onMethodCall(MessagePtr msg);
    void onEventSubscribe(MessagePtr msg);
    void onEventUnsubscribe(MessagePtr msg);
};

/******************************************************************************
 **    FUNCTION DEFINITIONS
 ******************************************************************************/
IpcServiceProvider::impl::impl(const std::string& srvName, uint32_t multiThreadCnt)
{
    /** todo something */
    mExecutor.setFunction(std::bind(&impl::onProcessMessage, this), multiThreadCnt);
    mService = new IpcService(srvName, [this](int32_t clientId, bool connected)
    {
         onClientStateChange(clientId, connected);
    });

    mParser = new IpcDataParser([this](MessagePtr msg)
    {
            mMsgQueue.push(msg); /** todo, refine me. */
    });
}

IpcServiceProvider::PandaErrc IpcServiceProvider::impl::start()
{
    PandaErrc ret = PandaErrc::NoInit;
    GuardAutoLock al(mApiLock);

    ret = mService->start([this](int32_t clientId, const uint8_t* data, uint32_t size)
        {
            ServiceClientMap::iterator it = mClients.find(clientId);

            if (it != mClients.end())
            {
                mParser->doParse(it->second.pktInfo, data, size);
            }
            
        });

    if (PandaErrc::OK == ret)
    {
        PandaErrc res = mExecutor.start();

        if (PandaErrc::OK == res)
        {
            mIsStarted = true;
        }
        else
        {
            LOGE("start function executor failed(%d, %s)\n", res, panda::core::ErrorStr(res));
            ret = PandaErrc::Undefined;
        }
    }

    return ret;
}

IpcServiceProvider::PandaErrc IpcServiceProvider::impl::stop()
{
    mIsTerminate = true;
    PandaErrc res = mExecutor.stop();
    if (PandaErrc::OK != res)
    {
        LOGE("stop function executor failed(%d, %s)\n", res, panda::core::ErrorStr(res));
        return PandaErrc::Undefined;
    }

    mService->stop();
    mIsStarted = false;
    return PandaErrc::OK;
}

uint32_t IpcServiceProvider::impl::generatePacketId()
{
    GuardAutoLock al(mGeneratePacketIdLock);
    uint32_t ret = ++mNextPacketId;

    if (ret == 0)
    {
        ret = ++mNextPacketId;
    }

    return ret;
}

IpcServiceProvider::PandaErrc IpcServiceProvider::impl::registerMethod(const std::string& name, IpcServiceMethodHandler handler)
{
    GuardAutoLock al(mApiLock);

    PandaErrc ret = PandaErrc::InvalidStatus;
    if (mIsStarted == false)
    {
        mMethods[name] = handler;
        ret = PandaErrc::OK;
    }
    else
    {
        LOGE("method need register before,api 'start' called\n");
    }

    return ret;
}

IpcServiceProvider::PandaErrc IpcServiceProvider::impl::sendResponse(IpcServiceRequestHandlePtr reqHdl, const uint8_t* data, uint32_t size)
{
    GuardAutoLock al(mApiLock);
    PandaErrc ret = PandaErrc::BadParameter;
    if (reqHdl)
    {
        if (mIsStarted)
        {
            ret = sendPacket(reqHdl->clientId, reqHdl->name, reqHdl->sessionId, MessageType::kResponse, data, size);
        }
        else
        {
            ret = PandaErrc::NoInit;
        }
    }

    return ret;
}

void IpcServiceProvider::impl::emitSignal(const std::string& sigName, const uint8_t* data, uint32_t size)
{
    if (mIsStarted)
    {
        GuardAutoLock gal(mEventLock);
        MessagePtr msg = MakeMessagePtr();
        msg->name = sigName;
        msg->type = MessageType::kEvent;
        msg->payloadSize = size;
        msg->payload = MakePayloadPtr(size);
        msg->sessionId = generatePacketId();
        memcpy(msg->payload.get(), data, size);


        mEventBackup[sigName] = msg; /** back up current event */

        ServiceEventMap::iterator it = mEvents.find(sigName);

        if (it != mEvents.end())
        {
            EventSubscribers& subs = it->second;
            for (auto clientId : subs)
            {
                 sendMessage(clientId, msg);
            }
        }
        
    }
}

IpcServiceProvider::PandaErrc IpcServiceProvider::impl::sendPacket(int32_t clientId, const std::string& name, uint32_t sessionId, MessageType type, const uint8_t* param, uint32_t size)
{
    IpcPacketParam ipp;
    ipp.name = name;
    ipp.sessionId = sessionId;
    ipp.type = type;
    IpcPacketData pktData = mParser->makePacket(ipp, param, size);

    return mService->sendData(clientId, pktData.payload.get(), pktData.size);
}

IpcServiceProvider::PandaErrc IpcServiceProvider::impl::sendMessage(int32_t clientId, MessagePtr msg)
{
    return sendPacket(clientId, msg->name, msg->sessionId, msg->type, msg->payload.get(), msg->payloadSize);
}

void IpcServiceProvider::impl::onClientStateChange(int32_t clientId, bool connected)
{
    GuardAutoLock gal(mClientLock);
    ServiceClientMap::iterator it = mClients.find(clientId);

    if (connected)
    {
        if (it == mClients.end())
        {
            IpcServiceClientInfo& cli = mClients[clientId];
            cli.clientId = clientId;
        }
        else
        {
            LOGW("clientId(%d) already exist, so do nothing\n", clientId);
        }
    }
    else
    {
        if (it != mClients.end())
        {
            {
                GuardAutoLock evtLock(mEventLock);
                for (auto evt : mEvents)
                {
                    EventSubscribers& subs = evt.second;
                    subs.erase(clientId);
                }
            }

            mClients.erase(it);
        }
        else
        {
            LOGW("clientId(%d) not found\n", clientId);
        }
    }
}

bool IpcServiceProvider::impl::onProcessMessage()
{
    MessagePtr msg = mMsgQueue.popAndWait();

    if (msg)
    {
        switch (msg->type)  /** todo refine, dispatch to independent thread to proccess */
        {
        case MessageType::kEventSubscribe:
            onEventUnsubscribe(msg);
            break;
        case MessageType::kEventUnsubscribe:
            onEventUnsubscribe(msg);
            break;
        case MessageType::kRequest:
            onMethodCall(msg);
            break;
        default:
            break;
        }
    }

    return !mIsTerminate;
}

void IpcServiceProvider::impl::onMethodCall(MessagePtr msg)
{
    const std::string& methodName = msg->name;

    ServiceMethodMap::iterator it = mMethods.find(methodName);

    if (it != mMethods.end())
    {
        IpcServiceMethodHandler method = it->second;
        if (method)
        {
            IpcServiceRequestHandlePtr srhdl = std::make_shared<IpcServiceRequestHandle>();
            srhdl->clientId = msg->clientId;
            srhdl->sessionId = msg->sessionId;
            srhdl->name = methodName;
            IpcPacketData pktData;
            pktData.payload = msg->payload;
            pktData.size = msg->payloadSize;
            method(srhdl, methodName, pktData);
        }
    }

}

void IpcServiceProvider::impl::onEventSubscribe(MessagePtr msg)
{
    GuardAutoLock gal(mEvtRegLock);

    const std::string& sigName = msg->name;
    int32_t clientId = msg->clientId;

    EventBackup::iterator it = mEventBackup.find(sigName);

    if (it != mEventBackup.end())
    {
        MessagePtr evt = MakeMessagePtr();

        *evt = *it->second;
        sendMessage(clientId, evt);
    }

    mEvents[sigName].emplace(clientId);
}

void IpcServiceProvider::impl::onEventUnsubscribe(MessagePtr msg)
{
    GuardAutoLock gal(mEvtRegLock);

    ServiceEventMap::iterator it = mEvents.find(msg->name);

    if (it != mEvents.end())
    {
        EventSubscribers& subs = it->second;
        subs.erase(msg->clientId);
    }
}

///////////////////////////////////////////////////////////////////////////////
///////////////////////////////////////////////////////////////////////////////
///////////////////////////////////////////////////////////////////////////////

IpcServiceProvider::IpcServiceProvider(const std::string& srvName, uint32_t multiThreadCnt /*= 1*/)
    :mImpl(new impl(srvName, multiThreadCnt))
{
    /** todo something */
}

IpcServiceProvider::~IpcServiceProvider()
{
    /** todo something */
}


IpcServiceProvider::PandaErrc IpcServiceProvider::start()
{
    return mImpl->start();
}

IpcServiceProvider::PandaErrc IpcServiceProvider::stop()
{
    return mImpl->stop();
}

IpcServiceProvider::PandaErrc IpcServiceProvider::registerMethod(const std::string& name, IpcServiceMethodHandler handler)
{
    return mImpl->registerMethod(name, handler);
}

IpcServiceProvider::PandaErrc IpcServiceProvider::sendResponse(IpcServiceRequestHandlePtr reqHdl, const uint8_t* data, uint32_t size)
{
    return mImpl->sendResponse(reqHdl, data, size);
}

void IpcServiceProvider::emitSignal(const std::string& sigName, const uint8_t* data, uint32_t size)
{
    return mImpl->emitSignal(sigName, data, size);
}


} /** namespace sys */
} /** namespace panda */


